package drds.binlog.parse.mysql;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.LifecycleAware;
import drds.binlog.binlog_event.LogByteBlock;
import drds.binlog.binlog_event.LogContext;
import drds.binlog.binlog_event.LogDecoder;
import drds.binlog.binlog_event.LogEvent;
import drds.binlog.binlog_event.event.rows_log_event.DeleteRowsLogEvent;
import drds.binlog.binlog_event.event.rows_log_event.UpdateRowsLogEvent;
import drds.binlog.binlog_event.event.rows_log_event.WriteRowsLogEvent;
import drds.binlog.common.Authors;
import drds.binlog.data_object.Entry;
import drds.binlog.parse.TableMetaData;
import drds.binlog.parse.exception.ParseException;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;

@Author(name = Authors.LI_YANG)
class SimpleParserStage implements EventHandler<MessageEvent>, LifecycleAware
{

    @Setter
    @Getter
    private MultiStageCoprocessor multiStageCoprocessor;
    @Setter
    @Getter
    private LogDecoder logDecoder;
    @Setter
    @Getter
    private LogContext logContext;

    public SimpleParserStage(MultiStageCoprocessor multiStageCoprocessor)
    {
        this.multiStageCoprocessor = multiStageCoprocessor;
        logDecoder = new LogDecoder(LogEvent.unknown_event, LogEvent.enum_end_event);
        logContext = new LogContext();
        if (multiStageCoprocessor.gtidSet != null)
        {
            logContext.setGtidSet(multiStageCoprocessor.gtidSet);
        }
    }

    public void onEvent(MessageEvent messageEvent, long sequence, boolean endOfBatch) throws Exception
    {
        try
        {
            LogEvent logEvent = messageEvent.getLogEvent();
            if (logEvent == null)
            {
                LogByteBlock logByteBlock = messageEvent.getLogByteBlock();
                logEvent = logDecoder.decode(logByteBlock, logContext);
                messageEvent.setLogEvent(logEvent);
            }

            int eventType = logEvent.getLogHeader().getLogType();
            TableMetaData tableMetaData = null;
            boolean needDmlParse = false;
            switch (eventType)
            {
                case LogEvent.write_rows_event_v1:
                case LogEvent.write_rows_event:
                    tableMetaData = multiStageCoprocessor.binlogLogEventConverter.parseRowsEventForTableMetaData((WriteRowsLogEvent) logEvent);
                    needDmlParse = true;
                    break;
                case LogEvent.update_rows_event_v1:
                case LogEvent.update_rows_event:
                    tableMetaData = multiStageCoprocessor.binlogLogEventConverter.parseRowsEventForTableMetaData((UpdateRowsLogEvent) logEvent);
                    needDmlParse = true;
                    break;
                case LogEvent.delete_rows_event_v1:
                case LogEvent.delete_rows_event:
                    tableMetaData = multiStageCoprocessor.binlogLogEventConverter.parseRowsEventForTableMetaData((DeleteRowsLogEvent) logEvent);
                    needDmlParse = true;
                    break;
                case LogEvent.rows_query_log_event:
                    needDmlParse = true;
                    break;
                default:
                    Entry entry = multiStageCoprocessor.binlogLogEventConverter.parse(messageEvent.getLogEvent(), false);
                    messageEvent.setEntry(entry);
            }

            // 记录一下DML的表结构
            messageEvent.setNeedDmlParse(needDmlParse);
            messageEvent.setTableMetaData(tableMetaData);
        } catch (Throwable e)
        {
            multiStageCoprocessor.parseException = new ParseException(e);
            throw multiStageCoprocessor.parseException;
        }
    }

    @Override
    public void onStart()
    {

    }

    @Override
    public void onShutdown()
    {

    }
}
